package MyThreadPool;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 自定义线程池，实现ThreadPool接口
 * @author Abel
 *
 */
public class MyThreadPool implements ThreadPool{
	
	/* 计算机CPU数 */
	private static final int CPU_NUMS = Runtime.getRuntime().availableProcessors();
	/* 最大线程数，CPU数 + 1 */
	private static final int MAX_WORK_THREAD = CPU_NUMS + 1;
	/* 任务队列，final修饰，不可改变 */
	private final LinkedList<Runnable> taskQueue = new LinkedList<>();
	/* 工作线程集合，final修饰，不可改变*/
	private final HashSet<Worker> workers = new HashSet<>();
	/* 线程池是否关闭标识 */
	private volatile boolean isShutdown = false;

	public MyThreadPool() {
		for(int i=0;i<MAX_WORK_THREAD;i++) {
			Worker work = new Worker();
			workers.add(work);
			new Thread(work).start();
		}
	}
	public MyThreadPool(int num) {
		if(num <= 0 || num > MAX_WORK_THREAD) {
			num = MAX_WORK_THREAD;
		}
		for(int i=0;i<num;i++) {
			Worker work = new Worker();
			workers.add(work);
			new Thread(work).start();
		}
	}
	//工作线程
	class Worker implements Runnable{
		//线程是否需要存活标识
		private volatile boolean running = true;
		@Override
		public void run() {
			while(running) {
				Runnable task = null;
				synchronized(taskQueue) {
					while(taskQueue.isEmpty() && running) {
						try {
							//任务队列为空，阻塞等待
							taskQueue.wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
							Thread.currentThread().interrupt();
							return;
						}
					}
					if(!taskQueue.isEmpty()) {
						task = taskQueue.poll();
					}
				}
				if(task != null) {
					task.run();
				}
			}
			//如果线程已经被关闭，但是队列还有任务，执行完
			synchronized(taskQueue) {
				while(!running && !taskQueue.isEmpty()) {
					taskQueue.poll().run();
				}
			}
		}
		public void shutDown() {
			this.running = false;
		}
	}
	
	@Override
	public void execute(Runnable runnable) {
		//线程池未关闭
		if(!isShutdown) {
			//任务不为空，并且线程池未关闭
			if (runnable != null) {
				synchronized (taskQueue) {
					// 加入工作线程队列
					taskQueue.add(runnable);

					// 尝试唤醒线程
					taskQueue.notify();
				}
			} else {
				System.out.println("提交任务不能为空...");
			}
		//线程池已关闭
		} else {
			//拒绝提交的任务
			throw new RejectedExecutionException();
		}

	}
	@Override
	public void shutdown() {
		isShutdown = true;
		// 线程关闭循环
        for (Worker w : workers) {
            w.shutDown();
        }
        // 全部唤醒
        synchronized (taskQueue) {
        	taskQueue.notifyAll();
        }
	}
	@Override
	public boolean isShutdown() {
		return isShutdown;
	}
	@Override
	public int getMaxSize() {
		return MAX_WORK_THREAD;
	}
	@Override
	public int getQueueSize() {
		int size = 0;
		synchronized(taskQueue) {
			size = taskQueue.size();
		}
		return size;
	}
	@Override
	public int getActiveCount() {
		int size = 0;
		synchronized(workers) {
			size = workers.size();
		}
		return size;
	}
}